#ifndef __M_BROKER_H__
#define __M_BROKER_H__

#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include "mq_connection.hpp"
#include "mq_virtualhost.hpp"
#include "mq_consumer.hpp"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_threadpool.hpp"
#include <stdio.h>
#include <unistd.h>

namespace mq{
using namespace std::placeholders;

class BrokerServer
{
    #define DEF_FILE "df_file"
    public:
        BrokerServer(const uint16_t& port,const std::string& basedir, const std::string& filename = DEF_FILE)
        :_server(&_loop, muduo::net::InetAddress("0.0.0.0",port),"MyTcpServer", muduo::net::TcpServer::kReusePort),
        _dispatcher(std::bind(&BrokerServer::onUnknownMessage, this,_1,_2,_3)),
        _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, _1, _2, _3))),
        _vHost(std::make_shared<Virtualhost>("host1",basedir,basedir + filename)),
        _connManager(std::make_shared<ConnectionManager>()),
        _pool(std::make_shared<ThreadPool>()),
        _consumerManager(std::make_shared<ConsumerManager>())
        {
            _dispatcher.registerMessageCallback<mq::openChannelRequest>( 
                std::bind(&BrokerServer::onOpenChannel, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mq::closeChannelRequest>( 
                std::bind(&BrokerServer::onCloseChannel, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mq::declareExchangeRequest>( 
                std::bind(&BrokerServer::onDeclareExchange, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mq::deleteExchangeRequest>( 
                std::bind(&BrokerServer::onDeleteExchange, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mq::declareQueueRequest>( 
                std::bind(&BrokerServer::onDeclareQueue, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mq::deleteQueueRequest>( 
                std::bind(&BrokerServer::onDeleteQueue, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mq::queueBindRequest>( 
                std::bind(&BrokerServer::onQueueBind, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mq::queueUnBindRequest>( 
                std::bind(&BrokerServer::onQueueUnBind, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mq::basicPublishRequest>( 
                std::bind(&BrokerServer::onBasicPublish, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mq::basicAckRequest>( 
                std::bind(&BrokerServer::onBasicAck, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mq::basicConsumerRequest>( 
                std::bind(&BrokerServer::onBasicConsumer, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mq::basicCancelRequest>( 
                std::bind(&BrokerServer::onBasicCancel, this, _1, _2, _3));

            _server.setConnectionCallback(
                std::bind(&BrokerServer::onConnection, this, _1));
            _server.setMessageCallback(
                std::bind(&ProtobufCodec::onMessage, _codec.get(), _1, _2, _3));

        }

        void start()
        {
            _server.start();
            _loop.loop();
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if(conn->connected()){
                _connManager->openConnection(_consumerManager,conn,_codec,_vHost,_pool);
            }else{
                _connManager->closeConnection(conn);
            }
        }

        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn,
                                const MessagePtr& message,
                                muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }
        void onOpenChannel(const muduo::net::TcpConnectionPtr& conn,
                                const openChannelRequestPtr& req,
                                muduo::Timestamp) 
        {
            Connection::ptr cp = _connManager->getConnection(conn);
            if(cp.get() == nullptr){
                LOG_ERRORM("openChannel false... no find Connection");
                conn->shutdown();
            }
            cp->openChannel(req);
        }
        
        void onCloseChannel(const muduo::net::TcpConnectionPtr& conn,
                                const closeChannelRequestPtr& req,
                                muduo::Timestamp)
        {
            Connection::ptr cp = _connManager->getConnection(conn);
            if(cp.get() == nullptr){
                LOG_ERRORM("closeChannel false... no find Connection");
                conn->shutdown();
            }
            cp->closeChannel(req);
        }

        void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn,
                                const declareExchangeRequestPtr& req,
                                muduo::Timestamp)
        {
            Connection::ptr cp = _connManager->getConnection(conn);
            if(cp.get() == nullptr){
                LOG_ERRORM("no find Connection");
                conn->shutdown();
            }
            Channel::ptr cnp = cp->getChannel(req->cid());
            if(cnp.get() == nullptr){
                LOG_ERRORM("no find Channel");
                conn->shutdown();
            }
            cnp->declareExchange(req);
        }
        void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn,
                                const deleteExchangeRequestPtr& req,
                                muduo::Timestamp)
        {
            Connection::ptr cp = _connManager->getConnection(conn);
            if(cp.get() == nullptr){
                LOG_ERRORM("no find Connection");
                conn->shutdown();
            }
            Channel::ptr cnp = cp->getChannel(req->cid());
            if(cnp.get() == nullptr){
                LOG_ERRORM("no find Channel");
                conn->shutdown();
            }
            cnp->deleteExchange(req);
        }
        void onDeclareQueue(const muduo::net::TcpConnectionPtr& conn,
                                const declareQueueRequestPtr& req,
                                muduo::Timestamp)
        {
            Connection::ptr cp = _connManager->getConnection(conn);
            if(cp.get() == nullptr){
                LOG_ERRORM("no find Connection");
                conn->shutdown();
            }
            Channel::ptr cnp = cp->getChannel(req->cid());
            if(cnp.get() == nullptr){
                LOG_ERRORM("no find Channel");
                conn->shutdown();
            }
            cnp->declareQueue(req);            
        }
        void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn,
                                const deleteQueueRequestPtr& req,
                                muduo::Timestamp)
        {
            Connection::ptr cp = _connManager->getConnection(conn);
            if(cp.get() == nullptr){
                LOG_ERRORM("no find Connection");
                conn->shutdown();
            }
            Channel::ptr cnp = cp->getChannel(req->cid());
            if(cnp.get() == nullptr){
                LOG_ERRORM("no find Channel");
                conn->shutdown();
            }
            cnp->deleteQueue(req);            
        }
        void onQueueBind(const muduo::net::TcpConnectionPtr& conn,
                                const queueBindRequestPtr& req,
                                muduo::Timestamp)
        {
            Connection::ptr cp = _connManager->getConnection(conn);
            if(cp.get() == nullptr){
                LOG_ERRORM("no find Connection");
                conn->shutdown();
            }
            Channel::ptr cnp = cp->getChannel(req->cid());
            if(cnp.get() == nullptr){
                LOG_ERRORM("no find Channel");
                conn->shutdown();
            }
            cnp->bind(req);
        }
        void onQueueUnBind(const muduo::net::TcpConnectionPtr& conn,
                                const queueUnBindRequestPtr& req,
                                muduo::Timestamp)
        {
            Connection::ptr cp = _connManager->getConnection(conn);
            if(cp.get() == nullptr){
                LOG_ERRORM("no find Connection");
                conn->shutdown();
            }
            Channel::ptr cnp = cp->getChannel(req->cid());
            if(cnp.get() == nullptr){
                LOG_ERRORM("no find Channel");
                conn->shutdown();
            }
            cnp->unBind(req);
        }
        void onBasicPublish(const muduo::net::TcpConnectionPtr& conn,
                                const basicPublishRequestPtr& req,
                                muduo::Timestamp)
        {
            Connection::ptr cp = _connManager->getConnection(conn);
            if(cp.get() == nullptr){
                LOG_ERRORM("no find Connection");
                conn->shutdown();
            }
            Channel::ptr cnp = cp->getChannel(req->cid());
            if(cnp.get() == nullptr){
                LOG_ERRORM("no find Channel");
                conn->shutdown();
            }            
            cnp->basicPublish(req);
        }
        void onBasicAck(const muduo::net::TcpConnectionPtr& conn,
                                const basicAckRequestPtr& req,
                                muduo::Timestamp)
        {
            Connection::ptr cp = _connManager->getConnection(conn);
            if(cp.get() == nullptr){
                LOG_ERRORM("no find Connection");
                conn->shutdown();
            }
            Channel::ptr cnp = cp->getChannel(req->cid());
            if(cnp.get() == nullptr){
                LOG_ERRORM("no find Channel");
                conn->shutdown();
            }          
            cnp->basicAck(req);  
        }
        void onBasicConsumer(const muduo::net::TcpConnectionPtr& conn,
                                const basicConsumerRequestPtr& req,
                                muduo::Timestamp)
        {
            Connection::ptr cp = _connManager->getConnection(conn);
            if(cp.get() == nullptr){
                LOG_ERRORM("no find Connection");
                conn->shutdown();
            }
            Channel::ptr cnp = cp->getChannel(req->cid());
            if(cnp.get() == nullptr){
                LOG_ERRORM("no find Channel");
                conn->shutdown();
            }            
            cnp->basicConsume(req);
        }
        void onBasicCancel(const muduo::net::TcpConnectionPtr& conn,
                                const basicCancelRequestPtr& req,
                                muduo::Timestamp)
        {
            Connection::ptr cp = _connManager->getConnection(conn);
            if(cp.get() == nullptr){
                LOG_ERRORM("no find Connection");
                conn->shutdown();
            }
            Channel::ptr cnp = cp->getChannel(req->cid());
            if(cnp.get() == nullptr){
                LOG_ERRORM("no find Channel");
                conn->shutdown();
            }           
            cnp->basicCancel(req); 
        } 
    private:
        muduo::net::EventLoop _loop;
        muduo::net::TcpServer _server;
        ProtobufDispatcher _dispatcher;
        ProtobufCodecPtr _codec;

        ConnectionManager::ptr _connManager;
        Virtualhost::ptr _vHost;
        ConsumerManager::ptr _consumerManager;
        ThreadPool::ptr _pool;
};

}

#endif